package com.sxzjrj.mblybc

import com.sxzjrj.config.MBYLBCConfig
import com.sxzjrj.constant.Constant
import com.sxzjrj.utils.JdbcUtils
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by ljj in 2019/3/23
  *
  */
object MBYLBCCount {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
      .setMaster("local[*]")
      .setAppName("慢病医疗报表")

    val sc = new SparkContext(conf)

    val result: RDD[((String, String), (Int, Double, Double, Double))] = sc.textFile("E:/BC_MBYLBC.txt")
      .map(_.split("\t", -1))
      .filter(arr => arr(1).length > 0 && arr(2).length > 0 && arr(4).length > 0 && arr.length >= Constant.DATALENGTH.toString.toInt)
      .map(arr => {

        val F_GRBM = arr(1) //个人编码
        val F_JZLX = arr(2) //就诊类型
        //val F_YLJG = arr(4) //医疗机构
        //val F_ID = if (arr(29).length > 0) arr(29) else "身份证信息缺失" //身份证号
        var F_SJ_JE = 0.0
        var F_HS_JE = 0.0
        var F_MB_YYF = 0.0
        var F_JZCS = 0
        try {
          F_SJ_JE = if (arr(41).length > 0) arr(41).toDouble else 0.0 //实际费用
          F_HS_JE = if (arr(42).length > 0) arr(42).toDouble else F_SJ_JE //核算金额
          F_MB_YYF = if (arr(43).length > 0) arr(43).toDouble else F_HS_JE //慢病总费用
          F_JZCS = if (F_SJ_JE == 0.0 && F_HS_JE == 0.0 && F_MB_YYF == 0.0) 0 else 1
        } catch {
          case e: Exception => e.printStackTrace()
        }

        ((F_GRBM, F_JZLX), (F_JZCS, F_SJ_JE, F_HS_JE, F_MB_YYF))

      }) //诊断次数
      .reduceByKey((a, b) => {
        (a._1 + b._1, a._2 + b._2, a._3 + b._3, a._4 + b._4)
      })



    JdbcUtils.saveData2Mysql(result)

//      .map(tp => ((tp._1._1,tp._1._2), tp._2._1))
//      .reduceByKey(_+_)
//      .foreach(println(_))


    sc.stop()


  }

}

